/*
 * =====================================================================================
 *       Filename:  factory.c
 *         Author:  MIEN
 *    Description:  队列工厂
 *           Date:  2015-07-10
 * =====================================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <dirent.h>

#include "lib/dict.h"
#include "config.h"
#include "queue.h"
#include "factory.h"
#include "dispatcher.h"

/* 队列工厂
 * 可自动创建不同名称的多个队列，并把数据送入对应的队列或从对应的队列取出。
 *
 * 在应用启动的时候:
 * factoryCreate();
 * factoryRestore();
 *
 * 在应用结束的时候:
 * factoryDump();
 * factoryFree();
 * 
 * 入队:
 * factoryIn(队列名，数据);
 *
 * 出队:
 * 数据 = factoryOut(队列名); */

static dictTable *_factory;
static dictType _factoryType;

static void _queueFree(void *val)
{
    queue *q = (queue*) val;
    queueFree(q);
}

void factoryCreate()
{
    _factoryType = dictTypeStringKeyFreeVal;
    _factoryType.valDel = _queueFree;
    _factory = dictCreate(&_factoryType);
}

void factoryFree()
{
    dictFree(_factory);
}

void factoryDump()
{
    int i;
    for (i = 0; i < _factory->size; i++) {
        dictNode *n = _factory->ht[i];
        while (n) {
            queue *q = (queue*) n->val;
            queueDump(q);
            n = n->next;
        }
    }
}

void factoryRestore()
{
    struct stat st;
    struct dirent *d;
    DIR *dir;
    char buf[256];

    char *data_dir = getConfig("queue_data_dir");
    dir = opendir(data_dir);
    if (dir == NULL) return;
    while ((d = readdir(dir)) != NULL) {
        if (d->d_name[0] == '.')
            continue;
        sprintf(buf, "%s/%s/_dump", data_dir, d->d_name);
        if (stat(buf, &st) != -1 && 
            (st.st_mode & S_IFMT) == S_IFREG) {
            queue *q = queueCreate(d->d_name);
            queueRestore(q);
            dictSet(_factory, d->d_name, q);
        }
    }
    closedir(dir);
}

void factoryIn(char *queue_name, char *data)
{
    queue *q = dictGet(_factory, queue_name);
    if (q == NULL) {
        q = queueCreate(queue_name);
        dictSet(_factory, queue_name, q);
    }
    queueIn(q, data);

    /* 通知推送 */
    if (q->suber != NULL) {
        notify(q);
    }
}

char *factoryOut(char *queue_name)
{
    queue *q = dictGet(_factory, queue_name);
    if (!q) return NULL;
    return queueOut(q);
}

int factorySub(char *queue_name, char *url, int number, int interval)
{
    queue *q = dictGet(_factory, queue_name);
    if (!q) return -1;
    return queueSub(q, url, number, interval);
}

int factoryUnsub(char *queue_name, char *url)
{
    queue *q = dictGet(_factory, queue_name);
    if (!q) return -1;
    return queueUnsub(q, url);
}

queue *factoryGetQueue(char *queue_name)
{
    return dictGet(_factory, queue_name);
}

void factoryAddQueue(char *queue_name)
{
    queue *q = queueCreate(queue_name);
    dictSet(_factory, queue_name, q);
}

void factoryDelQueue(char *queue_name)
{
    dictDel(_factory, queue_name);
}

dictTable *factoryGetQueuesTable()
{
    return _factory;
}

